package cn.siriusbot.rpc.client.processor;

import cn.siriusbot.rpc.*;
import cn.siriusbot.rpc.entity.RpcBody;
import cn.siriusbot.rpc.entity.RpcError;
import cn.siriusbot.rpc.entity.RpcNativeBody;
import cn.siriusbot.rpc.utils.SerializeUtils;
import cn.siriusbot.siriuspro.error.MsgException;
import cn.siriusbot.siriuspro.web.websocket.surface.WebsocketSession;
import com.alibaba.fastjson2.JSONObject;
import lombok.SneakyThrows;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于ws连接的客户端处理器
 */
public class WebsocketRpcProcessor implements RpcProcessor, SiriusLoggerTransmit {

    private final Map<String, RpcBody> rpcBodyMap = new ConcurrentHashMap<>();

    /**
     * session会话
     */
    private final WebsocketSession session;

    /**
     * 超时时间
     */
    private final long timeOut;

    /**
     * 创建WebsocketRpcProcessor处理器
     *
     * @param session session会话
     * @param timeOut 超时时间(ms)
     */
    public WebsocketRpcProcessor(WebsocketSession session, long timeOut) {
        this.session = session;
        this.timeOut = timeOut;
        log.info("已连接ws服务器 -> session_id = " + session.getId());
    }

    /**
     * 获取一个处理结果
     *
     * @param context 请求上下文
     * @param body    请求详细
     * @return 返回结果
     */
    @SneakyThrows
    @Override
    public Object getResult(RpcContext context, RpcNativeBody body) {
        final RpcBody rpcBody = context.toJsonRpcBody(body);
        rpcBodyMap.put(rpcBody.getId(), rpcBody);
        log.info("[WS]RPC请求 -> " + this.session.getId() + " body -> " + rpcBody);
        // 发送请求体
        session.send(JSONObject.toJSONString(rpcBody));
        // 堵塞等待
        synchronized (rpcBodyMap.get(rpcBody.getId())) {
            try {
                rpcBody.wait(timeOut);
                if (!rpcBody.isHandle()) {
                    throw new MsgException(500, "RPC处理超时!");
                }
                final RpcError error = rpcBody.getError();
                if (error != null) {
                    throw new MsgException(500, String.format("RPC服务器抛出异常，异常标记码:%s，异常信息:%s", error.getErrorId(), error.getMessage()));
                }
                // 处理返回结果
                return SerializeUtils.parsecJsonObject(rpcBody.getResult(), body.getReturnType());
            } finally {
                rpcBodyMap.remove(rpcBody.getId());
            }
        }
    }

    /**
     * 推入一个消息
     *
     * @param message 消息
     */
    public void putMessage(String message) {
        try {
            final RpcBody rpcBody = JSONObject.parseObject(message, RpcBody.class);
            final RpcBody sourceBody = rpcBodyMap.get(rpcBody.getId());
            if (sourceBody == null) {
                return;
            }
            // 设置结果
            sourceBody.setResult(rpcBody.getResult());
            sourceBody.setError(rpcBody.getError());
            sourceBody.setHandle(true);
            // 唤醒线程
            synchronized (rpcBodyMap.get(rpcBody.getId())) {
                sourceBody.notify();
            }
        } catch (Throwable ignored) {

        }
    }

    private SiriusLogger log = new PrintLog();

    @Override
    public SiriusLogger getLog() {
        return log;
    }

    @Override
    public void setLog(SiriusLogger logger) {
        this.log = logger;
    }
}
